/**
 * Project - The Three Phase Commit Protocol
 * Authors - Namrata Baliga, Aditya Joshi
 **/
import java.util.*;
import java.util.concurrent.*;

import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException; 
import java.io.PrintWriter;
import java.io.InputStreamReader;
import java.util.Scanner;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.StringTokenizer;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.net.SocketTimeoutException;
import java.net.Socket;
import java.net.ServerSocket;
import java.util.concurrent.ExecutionException;


// The Participant class encompasses the coordinator
// and non-coordinator roles.
class Participant
{
    int port,pid;
    String logFile;
    // ArrayList of processes a node believes to be operational
    ArrayList<Integer> upSet;
    //Current state of the process. Commit/Abort/Uncertain
    String state;
    ServerSocket pSocket;
    // monitor to make sure we have got responses from all nodes
    int allResp;
    boolean done;
    String currentSate;
    int numParticipants; 
    
    synchronized public void incrementcurrNumResp(String resp)
    {
        allResp++;
        if (resp.equals(Const.VOTE_NO))
        {
            System.out.println("no response");
            done = true;
            currentSate = resp;
            notifyAll();
        }
        else
        {
             System.out.println("all response");

            if (allResp == numParticipants)
            {
                done = true; 
                notifyAll();
            }
        }
    }

    // Creating a new node

    public Participant(int pid, int port)
    {
        this.pid = pid;
        this.port = port;
        // 0 to indicate responses not got from all
        this.allResp=0;
        // false  that we r not doen yet
        this.done = false;
        this.numParticipants = 0;
        this.currentSate = Const.VOTE_YES;
        // Log file to hold protocol state in stable storage
        // Structure: state;upset. eg: ABORT;1:2000,2:3000
        this.logFile = ""+pid+".log";
        this.upSet = new ArrayList<Integer>();
        this.state = "";
        System.out.println("PID: "+pid+" - Created new node at port: "+port);
    }

    // reading the log at recovery
    public void readLog()
    {
        System.out.println("DEBUG: "+pid+" - Reading log information ");
        try
        {
            // Initialize upset to 0's. At time of recovery the upSet volatile
            // information is lost.
            for (int j=0;j<10;j++)
            {
                upSet.add(0);
            }
            BufferedReader br = new BufferedReader(new FileReader(logFile));
            String lastLine="",line="";
            // Go till the last "not null" line of the logfile
            do
            {
                lastLine = line;
                line = br.readLine();
            }while (line!=null);
            System.out.println("DEBUG: "+pid+" - last log line"+lastLine);
            // The logfile splitter
            StringTokenizer logTokens = new StringTokenizer(lastLine, ";");
            // updating the state
            state = logTokens.nextToken();
            System.out.println("DEBUG: "+pid+" - state from log file"+state);
            // getting the upset
            StringTokenizer upSetTokens =  new StringTokenizer(logTokens.nextToken(), ",");
            while (upSetTokens.hasMoreTokens())
            {
                StringTokenizer pidPort = new StringTokenizer(upSetTokens.nextToken(), ":");
                upSet.set(Integer.parseInt(pidPort.nextToken()),
                        Integer.parseInt(pidPort.nextToken()));
            }
            System.out.println("DEBUG: "+pid+" - Done reading log information ");
        }
        catch(IOException io)
        {
            System.out.println("PID: "+pid+" - Caught io exception: "+io.getMessage());
        }

    }

    public String upSetListToString()
    {
        Integer upSetArr [] =  upSet.toArray(new Integer [upSet.size ()]);
        String upSetConcat="";
        // creating a upSet string that is comma separated
        // concatenation of all pid:port pairs
        for (int i=0;i<upSetArr.length;i++)
        {
            if (upSetArr[i]!=0)
            {
                if (upSetConcat.equals(""))
                {
                    upSetConcat+=i+":"+upSetArr[i];
                }
                else
                {
                    upSetConcat+=","+i+":"+upSetArr[i];
                }
            }    
        }
        return upSetConcat;
    }

    public ArrayList<Integer> upSetStrToList()
    {
        return null;
    }
    // Writing to the log. Synchronized to handle threads
    public synchronized void writeLog(String state,ArrayList<Integer> upSet)
    {
        System.out.println("DEBUG: "+pid+" - writing to log");
        try
        {
            // Writing to a log file
            BufferedWriter bw = new BufferedWriter(new FileWriter(logFile));
            
            // write to file
            bw.write(state+";"+upSetListToString());
            bw.newLine();
            bw.flush();
            bw.close();
            System.out.println("DEBUG: "+pid+" - Wrote to log: "+state+";"+upSetListToString());
            System.out.println("PID: "+pid+" - Saved process status to log");
        }
        catch(IOException io)
        {
            System.out.println("PID: "+pid+" - Caught io: "+io.getMessage());
        }
    }

    public void invokeParticipant()
    {
        // Invoke participant logic
        pSocket=null;
        ExecutorService threadExecutor = null;
        int retVal = 0;
        System.out.println("DEBUG: "+pid+" - invoking participant code"); 
        try
        {
            // Open sockect to listen for incoming requests
            pSocket = new ServerSocket(this.port);
            threadExecutor = Executors.newSingleThreadExecutor();
       }
        catch(IOException io)
        {
            System.out.println("PID: "+pid+" - Caught exception in sockect creation at port: "+this.port+". "+io.getMessage());
        }
        try
        {
            for (;;)
            {
                // Submit a participant thread. 
                System.out.println("submitting1");
                pSocket.setSoTimeout(Const.TIMEOUT);
                FutureTask<Integer> task = new FutureTask<Integer> (new ThreadParticipant(this, pSocket.accept()));
                threadExecutor.submit(task);
                retVal = task.get();
                System.out.println("DEBUG: "+pid+" val ret: "+retVal);
            }
        }
        catch(ExecutionException re)
        {
            System.out.println("DEBUG: "+pid+" - Caught io: " +re.getMessage());
            if ((re.getMessage()).indexOf(Const.UR_ELECTED) != -1)
            {
                // Coordinator has died and participant is new coordinator
                
                
            }
            else
            {
                // Process sent a 'no' response. ABORT sequence to be started.
                // write abort to log
                writeLog("ABORT",upSet);
           }
            threadExecutor.shutdown();
        }
        catch(InterruptedException re)
        {
            System.out.println("PID: "+pid+" Caught io: "+re.getMessage());
            threadExecutor.shutdown();
        }
        catch(SocketTimeoutException st)
        {
            // coordinator timeout. assumed died.
            //  election protocol
            threadExecutor.shutdown();
            System.out.println("Caught: "+st.getMessage());

        }
        catch(IOException io)
        {
            // coordinator died
            threadExecutor.shutdown();
            System.out.println("Caught: "+io.getMessage());
        }

        
    }

    public void invokeCoordinator(ArrayList<Integer> otherHosts)
    {
        // Coordinator action
        // Proactively connect to all participants
        System.out.println("Started coord");
        synchronized(this)
        {
            upSet.addAll(otherHosts);
        }
        writeLog("START",upSet);
        int numParticipants=0;
        ThreadCoordinator[] tc = new ThreadCoordinator[10];
        ExecutorService threadExecutor = null;
        threadExecutor = Executors.newFixedThreadPool(10);
        Set<Future<Integer>> set = new HashSet<Future<Integer>>();
        //        try
  //      {
            for (int i = 0;i<upSet.size();i++)
            {
                System.out.println("Invoked for node: "+i+" upSet:"+upSet.get(i));
                if (upSet.get(i)!=0)
                {
                    numParticipants++;
                    FutureTask<Integer> task = new 
                        FutureTask<Integer> (new ThreadCoordinator(this,i, upSet.get(i)));
                    System.out.println("Invoked for node: "+i);
                    threadExecutor.submit(task);
                    set.add(task);
                    //int retVal = task.get();
                }   
            }
            int retVal =0;
        try
        {
             for (Future<Integer> task : set)
                retVal = task.get();
       }
        catch(ExecutionException io)
        {
            threadExecutor.shutdown();
            System.out.println("only after finish");
        } 
        catch(InterruptedException io)
        {
        }
    }

    public void recover()
    {
        System.out.println("PID: "+pid+" - Recovering from crash ");
        // recover from log file
        readLog();       
        // reach a decision
        terminate();
    }

    public void terminate()
    {
        System.out.println("PID: "+pid+" - Initiating the termination protocol");
    }

    public void cleanUp()
    {
        System.out.println("PID: "+pid+" - Cleaning up process information");
        File f = new File(logFile);
        if (f.exists())
        {
            f.delete();
        }
        System.out.println("PID: "+pid+" - Clean up complete");
    }
}

class ThreadParticipant implements Callable 
{
    public int pid;
    public int port;
    public Socket cSocket;
    public Participant node;
    public ThreadParticipant(Participant node, Socket cSocket)
    {
        //this.pid = pid;
        //this.port = port;
        this.node = node;
        this.cSocket = cSocket;
    }

    public Integer call() throws Exception
    {
        // Participant waits for coordinator
        System.out.println("Waiting!!!");
        System.out.println("PID: "+node.pid+" - New incoming connection");
        // Set i/o streams
        PrintWriter out = new PrintWriter(cSocket.getOutputStream(), true);
        BufferedReader in = new BufferedReader(new InputStreamReader(
                    cSocket.getInputStream()));
        // read response
        String readData = in.readLine();
        System.out.println("PID: "+node.pid+" - Obtained "+readData);
        if (readData.indexOf(Const.VOTE_REQ) != -1)
        {
            Scanner stdin = new Scanner(System.in);
            System.out.print("PID: "+node.pid+" - Enter Vote (yes/no): ");
            String vote = stdin.nextLine();
            System.out.print("PID: "+node.pid+" - Sent Vote "+vote+" to coordinator");
            out.println(vote.toUpperCase());             
            if (vote.compareToIgnoreCase("YES") == 0)
            {
                System.out.println("PID: "+node.pid+" - Waiting for ABORT or PRECOMMIT ");
                while ((readData = in.readLine() ) == null)
                {
                }
                System.out.println("GOT: "+readData);
                if (readData.equals(Const.ABORT))
                out.println(Const.ACK);
                //in.readLine();
                System.out.println(in.readLine());
            }
            else
            {
                System.out.println("PID: "+node.pid+" - Initiating ABORT sequence");
                throw new ExecutionException("SENT NO",null);
            }
        }
        return 0;


 
    }

}
class ThreadCoordinator implements Callable 
{
    public int pid;
    public int port;
    public Participant node;
    public ThreadCoordinator(Participant node, int pid, int port)
    {
        this.pid = pid;
        this.port = port;
        this.node = node;
    }

    public Integer call() throws Exception
    {
        String sent="",recv="";
        System.out.println("PID: "+this.node.pid+" - Connecting to process: "+this.pid);
        // connecting to participant
        Socket cSocket = new Socket("localhost",this.port);
        System.out.println("PID: "+this.node.pid+" - Connected to process: "+this.pid);
        //cSocket.setSoTimeout(Const.TIMEOUT);
        //setting in/out buffers
        PrintWriter out = new PrintWriter(cSocket.getOutputStream(), true);
        BufferedReader in = new BufferedReader(new InputStreamReader(
                                           cSocket.getInputStream()));
        System.out.println("PID: "+this.node.pid+" - Sending vote request to all participants");
        out.println(Const.VOTE_REQ+";"+this.node.upSetListToString());
        recv = in.readLine();
        System.out.println("PID:"+this.node.pid+" - Got response "+recv 
                +"from participant: "+this.pid);
        this.node.incrementcurrNumResp(recv);
        System.out.println("node done: "+this.node.done);
        while (this.node.done != true)
        {   
            try
            {
                System.out.println("got stuck??");
                wait();
            }
            catch(InterruptedException ie)
            {
                System.out.println("awaken");

            }
        }
        if (this.node.currentSate.indexOf(Const.VOTE_NO) != -1)
        {
            // somebody sent a no
            this.node.writeLog("ABORT",null);
            out.println(Const.ABORT);

            throw new ExecutionException("GOT NO",null);
        }
        else
        {
            // all nodes sent a yes
            System.out.println("PID: "+this.node.pid+" - Sending PRECOMMIT");
            this.node.writeLog(Const.PRE_COMMIT,null);
            out.println(Const.PRE_COMMIT);
            recv = in.readLine();
            out.println(Const.COMMIT);
        } 
        return 0;
    }
}

class ThreePhaseCommit
{
    public static boolean isRecovering(String pid)
    {
        // Find out if the process is newly created or recovering
        boolean recover=false;
        File f = new File(pid+".log");
        if(f.exists())
            recover=true;
        return recover;
    }

    public static void main(String[] args)
    {
        // Obtain the id and port of the process
        System.out.println("First id-port: "+args[0]);
        StringTokenizer idPort = new StringTokenizer(args[0], ":");
        String pid = idPort.nextToken(); 
        Participant node = new Participant(Integer.parseInt(pid),Integer.parseInt(idPort.nextToken())); 
         // if exists DT log. recover from there.
        if (isRecovering(pid))
        {
            System.out.println("PID: "+pid+" - Recovering  from crash");
            node.recover();
        }
        else
        {
            System.out.println("New invocation");
            if (pid.equals("0"))
            {
                //Coordinator node
                // Obtain set of hosts from host port pairs
                System.out.println("PID: "+pid+" - Initializing Coordinator node");
                // initialize a 10 elelment arraylist with port '0'
                ArrayList<Integer> otherHosts = new ArrayList<Integer>();
                for (int j=0;j<10;j++)
                {
                    otherHosts.add(0);
                }
                // Parse arguments to obtain other nodes
                for (int i=1;i<args.length;i++)
                {
                    StringTokenizer otherIdPort = new StringTokenizer(args[i], ":");
                    String oPid = otherIdPort.nextToken();
                    otherHosts.set(Integer.parseInt(oPid),Integer.parseInt(otherIdPort.nextToken()));
                    System.out.println("E:"+otherHosts.get(Integer.parseInt(oPid)));
                }
                // Pass arraylist of othernodes and invoke coordinator
                node.invokeCoordinator(otherHosts);
                // Work done remove log
                node.cleanUp();
            }
            else
            {
                // The participant node
                System.out.println("PID: "+pid+" - Invoking Participant node");
                node.invokeParticipant();
                // work done remove log
                node.cleanUp();
            }
        }
    }
}
